/*
 * proxy_thread.cpp
 *
 *  Created on: 2013-12-19
 *      Author: lixingyi
 */

#include <errno.h>
#include "proxy_thread.h"
#include "network_util.h"
#include "time_util.h"
#include "system_util.h"

#include "mysql_thread.h"
#include "proxy_application.h"
#include "log_thread.h"


#define time_printf(fmt,args...) do{printf("%s:%d,%ld," fmt,__FILE__,__LINE__,time_ms(),##args) ; }while(0)

using namespace kingnet ;

enum
{
    PROXY_STATUS_CONNECT = 0 ,
    PROXY_STATUS_SEND_HANDSHAKE ,
    PROXY_STATUS_READ_AUTH ,
    PROXY_STATUS_SEND_AUTH_RESULT ,
    PROXY_STATUS_READ_QUERY ,
    PROXY_STATUS_WAIT_QUERY_RESULT ,
    PROXY_STATUS_SEND_RESULT ,
    PROXY_STATUS_SEND_COLUMN ,
    PROXY_STATUS_SEND_EOF ,
    PROXY_STATUS_SEND_ROW ,
    PROXY_STATUS_SEND_FIELD ,
    PROXY_STATUS_SEND_END ,
} ;


ProxyThread::ProxyThread():m_drizzle(NULL),m_client(NULL),
    m_response(NULL),m_row(NULL),m_status(PROXY_STATUS_CONNECT)
{
    // TODO Auto-generated constructor stub

    pthread_mutex_init(&m_mutex,NULL);
    pthread_cond_init(&m_cond,NULL);
}

ProxyThread::~ProxyThread()
{
    // TODO Auto-generated destructor stub

    pthread_cond_destroy(&m_cond); 
    pthread_mutex_destroy(&m_mutex);
}



int ProxyThread::on_init()
{

    //m_response_queue.init(1024) ;

    m_drizzle = drizzle_create() ;
    if(m_drizzle == NULL ) error_return(-1,"init drizzle failed") ;

    drizzle_set_context(m_drizzle,this) ;
    drizzle_set_timeout(m_drizzle,2000) ;
    drizzle_set_option(m_drizzle,DRIZZLE_FREE_OBJECTS,true) ;
    drizzle_set_option(m_drizzle, DRIZZLE_NON_BLOCKING, true);

    drizzle_con_st *con= drizzle_con_create(m_drizzle);
    if(con == NULL )
    {
        close(m_client_fd) ;
        drizzle_free(m_drizzle) ;
        return -1;
    }

    drizzle_con_set_fd(con, m_client_fd);
    drizzle_con_add_options(con, DRIZZLE_CON_MYSQL);

    //handshake
    //drizzle_con_set_auth(con, from->user, NULL);
    //drizzle_con_set_scramble(con, from->scramble);
    //drizzle_con_set_db(con, from->schema);
    drizzle_con_set_protocol_version(con, 0x0a);
    drizzle_con_set_server_version(con, "proxy_server");
    drizzle_con_set_thread_id(con, 1);
    drizzle_con_set_scramble(con, (const uint8_t *)"ABCDEFGHIJKLMNOPQRST");
    drizzle_con_set_capabilities(con, DRIZZLE_CAPABILITIES_CLIENT);
    drizzle_con_set_charset(con,0x21); //utf8
    drizzle_con_set_status(con, DRIZZLE_CON_STATUS_NONE);
    drizzle_con_set_max_packet_size(con, DRIZZLE_MAX_PACKET_SIZE);
    m_client = con ;
    m_status = PROXY_STATUS_CONNECT ;
    

    regcomp(&m_re_role_id,"role_id *= *'?([0-9]+)'?",REG_ICASE|REG_EXTENDED);
    regcomp(&m_re_sid ,"sid *= *'?([0-9]+)'?",REG_ICASE|REG_EXTENDED);
    regcomp(&m_re_server_id ,"server_id *= *'?([0-9]+)'?",REG_ICASE|REG_EXTENDED);

    m_last_time = 0 ;
    set_thread_title("proxy_thread");
    thread_info_log(get_app().log_thread(),"proxy_thread [%d] started",m_client_fd) ;
    return 0 ;
}

void ProxyThread::on_fini()
{
    
    thread_info_log(get_app().log_thread(),"proxy_thread [%d] stopped",m_client_fd) ;

    DrizzleResponse* response = NULL ;
    while(m_response_queue.pop(response) ==true  ) delete response ;
    if(m_response) delete m_response ;

    //m_response_queue.fini() ;

    drizzle_free(m_drizzle) ;

    regfree(&m_re_role_id) ;
    regfree(&m_re_sid) ;
    regfree(&m_re_server_id) ;

}


drizzle_return_t ProxyThread::status_wait_query_result()
{
    if( m_response_queue.empty() )
    {
        struct timespec ts = {0} ;
        ts.tv_sec = time(0) + 2 ;
        //no shared status , so ignore mutex
        //pthread_mutex_lock(&m_mutex);
        pthread_cond_timedwait(&m_cond,&m_mutex,&ts) ;
        //pthread_mutex_unlock(&m_mutex);
    }

    DrizzleResponse* response = NULL ;
    if(m_response_queue.pop(response) !=true || response == NULL) return DRIZZLE_RETURN_OK;
    
    drizzle_result_st* source = response->result() ;
    drizzle_result_st* result = drizzle_result_create_with(m_client,&m_result) ;

    drizzle_result_set_info(result, source->info);
    result->error_code= source->error_code;
    drizzle_result_set_sqlstate(result, source->sqlstate);
    result->warning_count= source->warning_count;
    result->insert_id= source->insert_id;
    result->affected_rows= source->affected_rows;
    result->column_count= source->column_count;
    result->row_count= source->row_count;

    m_response = response ;
    
    if( drizzle_result_error_code(result) != 0 || 
        drizzle_result_column_count(result)==0  )
    {
        m_status = PROXY_STATUS_SEND_END;
       
    }
    else
    {
        
        m_status = PROXY_STATUS_SEND_RESULT ;
    }
    
    return DRIZZLE_RETURN_OK ;

}



void ProxyThread::run_once()
{
    int now_time = time(0) ;
    if(now_time - m_last_time > 10)
    {
        m_last_time = now_time ;
        on_timer() ;
    }

    drizzle_return_t ret = DRIZZLE_RETURN_OK ;
    
    int8_t old_status = m_status ;
    do
    {
        old_status = m_status ;   
        switch(old_status)
        {
        case PROXY_STATUS_CONNECT :
            ret = status_connect() ;
            break ;
        case PROXY_STATUS_READ_AUTH :
            ret = status_read_auth() ;
            break ;
        case PROXY_STATUS_SEND_AUTH_RESULT:
            ret = status_send_auth_result() ;
            break ;
        case PROXY_STATUS_READ_QUERY:
            ret = status_read_query() ;
            break ;
        case PROXY_STATUS_WAIT_QUERY_RESULT:
            ret = status_wait_query_result() ;
            break ;

        //optimize code to minimise loop
    case PROXY_STATUS_SEND_RESULT:
        ret = drizzle_result_write(m_client,&m_result,false) ;
        //run next state directly 
        if( ret == DRIZZLE_RETURN_OK )  m_status = PROXY_STATUS_SEND_COLUMN ;
        else break ;  
    case PROXY_STATUS_SEND_COLUMN:
        
        while (drizzle_column_st* column= drizzle_column_next(m_response->result() ) )
        {
            ret = drizzle_column_write(&m_result,column) ;
            if( ret != DRIZZLE_RETURN_OK ) break ;
            
        }
        drizzle_result_set_eof(&m_result, true);        
        m_status = PROXY_STATUS_SEND_EOF ; 
        //run next state directly
        //break ;
    case PROXY_STATUS_SEND_EOF:
        ret = drizzle_result_write(m_client,&m_result,false) ;
        if( ret == DRIZZLE_RETURN_OK )
        {
            m_status = PROXY_STATUS_SEND_ROW ;
        }
        else break ;
    case PROXY_STATUS_SEND_ROW:
        if(m_row == NULL ) m_row = drizzle_row_next(m_response->result()) ;

        if(m_row == NULL )
        {
            m_status = PROXY_STATUS_SEND_END ;
            break ;
        }
        else
        {
            size_t* field_sizes = drizzle_row_field_sizes(m_response->result() ) ;
            drizzle_result_calc_row_size(&m_result,m_row,field_sizes) ;
            ret = drizzle_row_write(&m_result) ;
            if( ret != DRIZZLE_RETURN_OK ) { break; } 

            m_field_index = 0 ;
            m_status = PROXY_STATUS_SEND_FIELD ;
            //run next state directly
            
        }
        //break ;
    case PROXY_STATUS_SEND_FIELD:
            for( ; m_field_index < drizzle_result_column_count(m_response->result() ) ; ++m_field_index)
            {
                size_t size = strlen(m_row[m_field_index]) ;
                ret = drizzle_field_write(&m_result,m_row[m_field_index],size,size) ;
                if( ret != DRIZZLE_RETURN_OK ) { break ; }

            }

            if(m_field_index == drizzle_result_column_count(m_response->result()) )
            {
                m_status = PROXY_STATUS_SEND_ROW ;
                m_row = NULL ;
            }

            break ;

    case PROXY_STATUS_SEND_END:
        ret = drizzle_result_write(m_client,&m_result,true) ;
        if( ret == DRIZZLE_RETURN_OK ) 
        {
            if(m_response)
            {
                delete m_response ;
                m_response = NULL ;
            }
            drizzle_result_free(&m_result) ;
            m_status = PROXY_STATUS_READ_QUERY;
        }

        break ;

    default:
            thread_warn_log(get_app().log_thread(),"invalid status") ;
            ret = DRIZZLE_RETURN_UNEXPECTED_DATA;
            
        } ;

        //time_printf("old_status:%d new_status:%d\n",old_status,m_status) ;

    } while(old_status != m_status );

    if(ret == DRIZZLE_RETURN_IO_WAIT)
    {
        drizzle_con_wait(m_drizzle);
    }
    else if ( ret != DRIZZLE_RETURN_OK)
    {
        thread_warn_log(get_app().log_thread(),"proxy_thread [%d] ret:%d error:%s",
            m_client_fd,ret,drizzle_error(m_drizzle) ) ;
        this->stop() ;
    }
    


    
}




drizzle_return_t ProxyThread::status_read_query()
{
    drizzle_command_t command ;
    size_t total =0;
    //size_t offset=0,size = 0;
    drizzle_return_t ret ;
    void* sql = drizzle_con_command_buffer(m_client,&command,&total,&ret) ;
    //void* sql = drizzle_con_command_read(m_client,&command,&offset,&size,&total,&ret) ;
    //printf("read_query offset:%ld size:%ld total:%ld sql:%p ret:%d\n",offset,size,total,sql,ret) ;
    if(ret != DRIZZLE_RETURN_OK )
    {
        return ret ;
    }

    if( sql == NULL) return DRIZZLE_RETURN_NULL_SIZE;

    class drizzle_command_guard
    {
    public:
        drizzle_command_guard(void* sql):m_sql(sql) { } ;
        ~drizzle_command_guard() { drizzle_con_command_buffer_free((uint8_t*)m_sql); } ;
    private:
        void* m_sql ;
    } ;

    drizzle_command_guard guard(sql) ;

    if(command == DRIZZLE_COMMAND_QUIT )
    {
        drizzle_result_create_with(m_client,&m_result) ;
        drizzle_result_set_error_code(&m_result,0) ;
        m_status = PROXY_STATUS_SEND_END ;
        this->stop() ;
        return DRIZZLE_RETURN_OK ;
    }

    if(command != DRIZZLE_COMMAND_QUERY)
    {
        drizzle_result_create_with(m_client,&m_result) ;
        int error_code = 1046 ;
        if(command == DRIZZLE_COMMAND_INIT_DB || command == DRIZZLE_COMMAND_PING )
        {
            error_code = 0 ;
        }

        drizzle_result_set_error_code(&m_result,error_code) ;

        m_status = PROXY_STATUS_SEND_END ;
        return DRIZZLE_RETURN_OK ;
    }

    DrizzleRequest* request = new DrizzleRequest(this->shared_from_this(),(const char*)sql,total,command) ;
    if(request == NULL) 
    { 
        thread_warn_log(get_app().log_thread(),"create request failed") ;
        return DRIZZLE_RETURN_INTERNAL_ERROR ;
    }
 
    MysqlThread* thread = get_app().get_mysql_thread(get_thread_id(request->sql()))  ;
    if(thread == NULL || (!thread->running()) )
    { 
        thread_warn_log(get_app().log_thread(),"mysql thread not running") ;
        delete request ;
        return DRIZZLE_RETURN_INTERNAL_ERROR ;
    }
    
   
    if( thread->push_request(request)!=true)
    {
        thread_warn_log(get_app().log_thread(),"request queue full") ;
        delete request ;
        return DRIZZLE_RETURN_INTERNAL_ERROR ;
            
    }
    else
    {
        m_status = PROXY_STATUS_WAIT_QUERY_RESULT ;
        return DRIZZLE_RETURN_OK ;
    }
        
    return DRIZZLE_RETURN_INTERNAL_ERROR ;

}


void ProxyThread::on_timer()
{

}


int ProxyThread::get_thread_id(const char* sql)
{
    enum key_type 
    {
        key_type_unknown = 0 ,
        key_type_role_id = 1 ,
        key_type_sid = 2 ,
    } ;

    key_type kt = key_type_unknown;
    int key = 0 ;
    regmatch_t match[2] ;
    if( regexec(&m_re_role_id,sql,2,match,0) == 0)
    {
        kt = key_type_role_id ;
    }
    else if( regexec(&m_re_sid,sql,2,match,0) == 0)
    {
        kt = key_type_sid ;
    }
    else if( regexec(&m_re_server_id,sql,2,match,0) == 0)
    {
        kt = key_type_sid ;
    }

    int str_size = match[1].rm_eo - match[1].rm_so ;
    
    if(kt != key_type_unknown && str_size > 0 && str_size < 32 )
    {
        char str[32] ;
        memcpy(str,sql + match[1].rm_so,str_size) ;
        str[str_size] = '\0' ;
        key = atol(str) ;
    }

    thread_debug_log(get_app().log_thread(),"proxy_thread [%d] key:%d sql:%s",m_client_fd,key,sql) ;
    if( kt == key_type_role_id && key > 0 ) return key / 1000000 ;
    else if ( key == key_type_sid && key >0 ) return key ;

    return 0 ;
    
}


drizzle_return_t ProxyThread::status_connect()
{

    drizzle_return_t ret = drizzle_handshake_server_write(m_client) ;
    if(ret == DRIZZLE_RETURN_OK) m_status = PROXY_STATUS_READ_AUTH ;
    return ret ;
}

drizzle_return_t ProxyThread::status_read_auth()
{
    drizzle_return_t ret = drizzle_handshake_client_read(m_client);
    if(ret == DRIZZLE_RETURN_OK) m_status = PROXY_STATUS_SEND_AUTH_RESULT ;
    return ret ;
 
}

drizzle_return_t ProxyThread::status_send_auth_result()
{
    drizzle_result_create_with(m_client,&m_result) ;
    drizzle_result_set_error_code(&m_result,0) ;
    drizzle_return_t ret = drizzle_result_write(m_client, &m_result, true);
    if(ret == DRIZZLE_RETURN_OK) 
    {
        drizzle_result_free(&m_result) ;
        m_status = PROXY_STATUS_READ_QUERY;
    }

    return ret ;
 
}


/*
drizzle_return_t ProxyThread::status_proxy()
{

    drizzle_return_t ret = DRIZZLE_RETURN_OK ;

    switch(m_status)
    {
    case PROXY_STATUS_CONNECT:
        ret = drizzle_handshake_server_write(m_client) ;
        if(ret == DRIZZLE_RETURN_OK) m_status = PROXY_STATUS_READ_AUTH ;
       
        break ;
    case PROXY_STATUS_READ_AUTH:
        ret = drizzle_handshake_client_read(m_client);
        if(ret == DRIZZLE_RETURN_OK) m_status = PROXY_STATUS_SEND_AUTH_RESULT ;
        
        break ;
    case PROXY_STATUS_SEND_AUTH_RESULT:
        drizzle_result_create_with(m_client,&m_result) ;
        drizzle_result_set_error_code(&m_result,0) ;
        ret = drizzle_result_write(m_client, &m_result, true);
        if(ret == DRIZZLE_RETURN_OK) 
        {
            drizzle_result_free(&m_result) ;
            m_status = PROXY_STATUS_READ_QUERY;
        }
        
        break ; 
   
    default:
        ;

    }
    
    
   return ret ;
}
*/








